package com.zst.amqp;

import com.rabbitmq.client.Channel;
import com.zst.amqp.entity.MessageDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;

import java.nio.charset.StandardCharsets;

/**
 * 消息监听器
 *
 * @author zst
 */
public class MessageListener implements ChannelAwareMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MessageListener.class);
    /**
     * MessageDispatcher，用于把收到的消息分发到对应消息类型的消费者中
     */
    private MessageDispatcher messageDispatcher;
    /**
     * MessageSender，用于在消息消费失败时把消息发送到重试队列中
     */
    private MessageSender messageSender;
    /**
     * 消息重试次数
     */
    private int maxRetry;
    /**
     * 死信交换器名称
     */
    private String deadExchangeName;

    private static final String RETRY_HEADER = "retryCount";

    public MessageListener(MessageSender messageSender, MessageDispatcher messageDispatcher, int maxRetry, String deadExchangeName) {
        this.messageDispatcher = messageDispatcher;
        this.maxRetry = maxRetry;
        this.messageSender = messageSender;
        this.deadExchangeName = deadExchangeName;
    }

    /**
     * 消息监听方法，当应用收到消息时会调用此方法
     * @param message
     * @param channel
     * @throws Exception
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        try {
            invoke(message);
            //消息消费成功，发送确认给消息队列
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            int retryCount = getMessageRetryCount(message);
            if (retryCount < maxRetry) {
                // 如果消息的重试次数小于最大重试次数，则让消息队列进入死信队列中
                retryCount++;
                message.getMessageProperties().getHeaders().put(RETRY_HEADER, String.valueOf(retryCount));
                messageSender.send(message, message.getMessageProperties().getReceivedRoutingKey(), this.deadExchangeName);
                // 原本的消息队列发送确认收到
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                // 如果消息重试次数超过最大重试次数，则发送确认给消息队列，本地不作处理直接丢弃
                logger.error("discard message:" + new String(message.getBody(), StandardCharsets.UTF_8));
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        }
    }

    /**
     * 从AMQP Message中提取出消息的内容转化为MessageDTO并消费消息
     * @param message
     * @throws Exception
     */
    private void invoke(Message message) throws Exception {
        String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);
        MessageDTO dto = MessageDTO.jsonToDTO(messageContent);
        if (dto == null) {
            throw new Exception("Cannot instantiate message");
        }

        messageDispatcher.dispatch(dto);
    }

    /**
     * 从message中获取重试次数
     * @param message
     * @return
     */
    private int getMessageRetryCount(Message message) {
        String retryCountStr = (String) message.getMessageProperties().getHeaders().get(RETRY_HEADER);
        Integer retryCount = null;
        if (retryCountStr == null) {
            return 0;
        } else {
            return Integer.valueOf(retryCountStr);
        }
    }
}
